#include "tcp_connection.h"

int handle_connection_closed(struct tcp_connection *conn)
{
    struct event_loop *eloop = conn->eloop;
    struct channel *chan = conn->chan;
    event_loop_remove_channel_event(eloop, chan->fd, chan);
    if (conn->connection_closed_callback != NULL)
    {
        conn->connection_closed_callback(conn);
    }
    return 0;
}

int handle_read(void *data)
{
    struct tcp_connection *conn = (struct tcp_connection *)data;
    struct buffer *input_buffer = conn->input_buffer;
    struct channel *chan = conn->chan;

    if (buffer_socket_read(input_buffer, chan->fd) > 0)
    {
        if (conn->message_callback != NULL)
        {
            conn->message_callback(input_buffer, conn);
        }
    }
    else
    {
        handle_connection_closed(conn);
    }
    return 0;
}

int handle_write(void *data)
{
    struct tcp_connection *conn = (struct tcp_connection *)data;
    struct event_loop *eloop = conn->eloop;
    if (eloop->owner_thread_id != pthread_self())
    {
        sprintf(stderr, "event_loop must in same thread\n");
        exit(EXIT_FAILURE);
    }

    struct buffer *output_buffer = conn->out_buffer;
    struct channel *chan = conn->chan;

    ssize_t nwrited = write(chan->fd, output_buffer->data + output_buffer->read_index,
                            buffer_readable_size(output_buffer));
    if (nwrited > 0)
    {
        output_buffer->read_index += nwrited;
        if (buffer_readable_size(output_buffer) == 0)
        {
            channel_write_event_disable(chan);
        }
        if (conn->write_completed_callback != NULL)
        {
            conn->write_completed_callback(conn);
        }
    }
    else
    {
        printf("handle_write for tcp connection %s\n", conn->name);
    }
}

struct tcp_connection *tcp_connection_new(int fd, struct event_loop *eloop,
                                          connection_completed_callback_t connection_completed_callback,
                                          message_callback_t message_callback,
                                          write_completed_callback_t write_completed_callback,
                                          connection_closed_callback_t connection_closed_callback)
{
    struct tcp_connection *conn = malloc(sizeof(struct tcp_connection));
    conn->write_completed_callback = write_completed_callback;
    conn->message_callback = message_callback;
    conn->connection_completed_callback = connection_completed_callback;
    conn->connection_closed_callback = connection_closed_callback;
    conn->eloop = eloop;
    conn->input_buffer = buffer_new();
    conn->out_buffer = buffer_new();

    char *buf = malloc(16);
    sprintf(buf, "connection-%d\0", fd);
    conn->name = buf;

    struct channel *chan = channel_new(fd, EVENT_READ, handle_read, handle_write, conn);
    conn->chan = chan;

    if (conn->connection_completed_callback != NULL)
    {
        conn->connection_completed_callback(conn);
    }
    event_loop_add_channel_event(conn->eloop, fd, conn->chan);
    return conn;
}

int tcp_connection_send_data(struct tcp_connection *conn, void *data, int size)
{
    size_t nwrited = 0;
    size_t nleft = size;
    int fault = 0;

    struct channel *chan = conn->chan;
    struct buffer *output_buffer = conn->out_buffer;

    if (!channel_write_event_is_enabled(chan) && buffer_readable_size(output_buffer) == 0)
    {
        nwrited = write(chan->fd, data, size);
        if (nwrited < 0)
        {
            nleft = nleft - nwrited;
        }
        else
        {
            nwrited = 0;
            if (errno != EWOULDBLOCK)
            {
                if (errno == EPIPE || errno == ECONNRESET)
                {
                    fault = 1;
                }
            }
        }
    }
    if (!fault && nleft > 0)
    {
        buffer_append(output_buffer, data + nwrited, nleft);
        if (!channel_write_event_is_enabled(chan))
        {
            channel_write_event_enable(chan);
        }
    }
    return nwrited;
}

int tcp_connection_send_buffer(struct tcp_connection *conn, struct buffer *buf)
{
    int size = buffer_readable_size(buf);
    int result = tcp_connection_send_data(conn, buf->data + buf->read_index, size);
    buf->read_index += size;
    return result;
}

void tcp_connection_shutdown(struct tcp_connection *conn)
{
    if (shutdown(conn->chan->fd, SHUT_WR) < 0)
    {
        printf("tcp_connection_shutdown failed, socket == %d", conn->chan->fd);
    }
}